package cn.ac.iie.di.ban.ycl.cleaner.sender;

import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import org.apache.avro.generic.GenericRecord;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.configuration2.Configuration;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 张明诚
 * @version V1.0
 * @date 2016-11-9 18:37:42
 * @desc
 */
public class MessageSender {

    private static final Logger LOG = LoggerFactory.getLogger(MessageSender.class);
    private final REConnection conn;
    private final String topic;
    RESendSessionBuilder builder;
    RESendSession session;
    private Configuration conf;

    public MessageSender(String nameServer, String topic, Configuration conf) {
        conn = new REConnection(nameServer);
        this.topic = topic;
        this.conf = conf;
    }


    public void sendMessage(REMessage messages) throws Exception {
        String[] columnNames = messages.colNames;
        REFieldType[] types = messages.types;
        int columnLength = columnNames.length;
        if (builder == null) {
            builder = (RESendSessionBuilder) conn.
                    getSendSessionBuilder(this.topic);
            for (int i = 0; i < columnNames.length; i++) {
                builder.addColumn(columnNames[i], types[i], true);
            }
        }

        if (session == null) {
            session = (RESendSession) builder.build();
            session.start();
        }
        session.setDocSchemaName(messages.docSchemaName);
        session.setUserDesc((HashMap<String, String>) messages.userDesc);
        List<Object[]> datas = messages.data;
        for (Object[] data : datas) {
            int dataLength = data.length;
            if (dataLength != columnLength) {
                LOG.info("columnNames:{}",Arrays.toString(columnNames));
                throw new Exception("data length not equal column length!");
            }
            for (int i = 0; i < columnLength; i++) {
                if (data[i] == null) {
                    continue;
                }
                try {
                    switch (types[i]) {
                        case String:
                            session.setString(columnNames[i], String.valueOf(data[i]));
                            break;
                        case Int:
                            session.setInt(columnNames[i], Integer.valueOf(data[i].toString()));
                            break;
//                        增加判断，若给垃圾队列发送数据则给it字段置为当前时间戳
                        case Long:
                            if ("it".equals(columnNames[i]) && this.topic.equals(conf.getString(Constants.RUBBISH_SEND_TOPIC)))
                                session.setLong(columnNames[i], System.currentTimeMillis() / 1000);
                            else
                                session.setLong(columnNames[i], Long.valueOf(data[i].toString()));
                            break;
                        case Float:
                            session.setFloat(columnNames[i], Float.valueOf(data[i].toString()));
                            break;
                        case Double:
                            session.setDouble(columnNames[i], Double.valueOf(data[i].toString()));
                            break;
                        case Binary:
                            session.setBinary(columnNames[i], data[i].toString().getBytes());
                            break;
                        case Doc:
                            session.setDoc(columnNames[i], (GenericRecord) data[i]);
                            break;
                        case Docs:
                            session.setDocs(columnNames[i], (GenericRecord[]) data[i]);
                            break;
                        case Binaries:
                            session.setBinaries(columnNames[i], (byte[][]) data[i]);
                            break;
                        case Doubles:
                            session.setDoubles(columnNames[i], (double[]) data[i]);
                            break;
                        case Floats:
                            session.setFloats(columnNames[i], (float[]) data[i]);
                            break;
                        case Longs:
                            session.setLongs(columnNames[i], (long[]) data[i]);
                            break;
                        case Ints:
                            session.setInts(columnNames[i], (int[]) data[i]);
                            break;
                        case Strings:
                            session.setStrings(columnNames[i], (String[]) data[i]);
                            break;
                        case Boolean:
                            session.setBoolean(columnNames[i], (boolean) data[i]);
                            break;
                        case Booleans:
                            session.setBooleans(columnNames[i], (boolean[]) data[i]);
                            break;
                        case Struct:
                            session.setStruct(columnNames[i], (Map) data[i]);
                            break;
                        case Structs:
                            session.setStructs(columnNames[i], (Map[]) data[i]);
                            break;
                        default:
                            LOG.error("There is not the type of " + types[i]);
                    }
                } catch (RESessionException | NumberFormatException ex) {
                    LOG.error("the data transfer error and the column name is {}\n,"
                                    + "index is {},data:{},columns:{}"
                                    + ExceptionUtils.getFullStackTrace(ex),
                            columnNames[i], i, Arrays.toString(data),
                            Arrays.toString(columnNames));
                }
            }
            try {
                session.add();
            } catch (RESessionException ex) {
                while (true) {
                    try {
                        session.flush();
                        break;
                    } catch (Exception e) {
                    }
                }
                session.add();
            }
        }
        session.flush();
    }


   /* public void sendMessage(REMessage messages) throws Exception {

        //先转换为列表，Arrays.asList()返回的是静态列表，无法添加/删除
        List<String> tmpColNames = new ArrayList<>(Arrays.asList(messages.colNames));
        List<REFieldType> tmpTypes = new ArrayList<>(Arrays.asList(messages.types));

        //屏蔽消息里 列名数组 和 类型数组 里的demd5字段：得到demd5字段索引，然后删除该字段
        int demd5Index = tmpColNames.indexOf("demd5");
        tmpColNames.remove(demd5Index);
        tmpTypes.remove(demd5Index);

        String[] columnNames = null;
        REFieldType[] types = null;

        try {
            //还原为数组
            columnNames = tmpColNames.toArray(new String[0]);
            types = tmpTypes.toArray(new REFieldType[0]);
        }catch (Exception e){
            LOG.error("transfer array error!");
        }

        int columnLength = columnNames.length;
        if (builder == null) {
            builder = (RESendSessionBuilder) conn.
                    getSendSessionBuilder(this.topic);
            for (int i = 0; i < columnNames.length; i++) {
                builder.addColumn(columnNames[i], types[i], true);
            }
        }

        if (session == null) {
            session = (RESendSession) builder.build();
            session.start();
        }
        session.setDocSchemaName(messages.docSchemaName);
        session.setUserDesc((HashMap<String, String>) messages.userDesc);
        List<Object[]> datas = messages.data;
        for (Object[] data : datas) {
            int dataLength = data.length;
            if (dataLength != columnLength) {
                LOG.warn("data length not equal column length!");
            }
//            for (int i = 0; i < columnLength; i++) {
            for (int i = 0; i < dataLength; i++) {
                if (data[i] == null || i == demd5Index) {
                    continue;
                }
                try {
                    switch (types[i]) {
                        case String:
                            session.setString(columnNames[i], String.valueOf(data[i]));
                            break;
                        case Int:
                            session.setInt(columnNames[i], Integer.valueOf(data[i].toString()));
                            break;
//                        增加判断，若给垃圾队列发送数据则给it字段置为当前时间戳
                        case Long:
                            if ("it".equals(columnNames[i]) && this.topic.equals(conf.getString(Constants.RUBBISH_SEND_TOPIC)))
                                session.setLong(columnNames[i], System.currentTimeMillis() / 1000);
                            else
                                session.setLong(columnNames[i], Long.valueOf(data[i].toString()));
                            break;
                        case Float:
                            session.setFloat(columnNames[i], Float.valueOf(data[i].toString()));
                            break;
                        case Double:
                            session.setDouble(columnNames[i], Double.valueOf(data[i].toString()));
                            break;
                        case Binary:
                            session.setBinary(columnNames[i], data[i].toString().getBytes());
                            break;
                        case Doc:
                            session.setDoc(columnNames[i], (GenericRecord) data[i]);
                            break;
                        case Docs:
                            session.setDocs(columnNames[i], (GenericRecord[]) data[i]);
                            break;
                        case Binaries:
                            session.setBinaries(columnNames[i], (byte[][]) data[i]);
                            break;
                        case Doubles:
                            session.setDoubles(columnNames[i], (double[]) data[i]);
                            break;
                        case Floats:
                            session.setFloats(columnNames[i], (float[]) data[i]);
                            break;
                        case Longs:
                            session.setLongs(columnNames[i], (long[]) data[i]);
                            break;
                        case Ints:
                            session.setInts(columnNames[i], (int[]) data[i]);
                            break;
                        case Strings:
                            session.setStrings(columnNames[i], (String[]) data[i]);
                            break;
                        case Boolean:
                            session.setBoolean(columnNames[i], (boolean) data[i]);
                            break;
                        case Booleans:
                            session.setBooleans(columnNames[i], (boolean[]) data[i]);
                            break;
                        case Struct:
                            session.setStruct(columnNames[i], (Map) data[i]);
                            break;
                        case Structs:
                            session.setStructs(columnNames[i], (Map[]) data[i]);
                            break;
                        default:
                            LOG.error("There is not the type of " + types[i]);
                    }
                } catch (RESessionException | NumberFormatException ex) {
                    LOG.error("the data transfer error and the column name is {}\n,"
                                    + "index is {},data:{},columns:{}"
                                    + ExceptionUtils.getFullStackTrace(ex),
                            columnNames[i], i, Arrays.toString(data),
                            Arrays.toString(columnNames));
                }
            }
            try {
                session.add();
            } catch (RESessionException ex) {
                while (true) {
                    try {
                        session.flush();
                        break;
                    } catch (Exception e) {
                    }
                }
                session.add();
            }
        }
        session.flush();
    }*/

    public void shutdown() {
        if (session != null) {
            try {
                session.shutdown();
            } catch (REConnectionException ex) {
                LOG.error("", ex);
            }
        }
    }
}
